/*-------------------------------------------------------------------------
 *
 * basebackup_lz4.c
 *	  基备份接收器实现 lz4 压缩。
 *
 * Portions Copyright (c) 2010-2022, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *	  src/backend/backup/basebackup_lz4.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#ifdef USE_LZ4
#include <lz4frame.h>
#endif

#include "backup/basebackup_sink.h"

#ifdef USE_LZ4

typedef struct bbsink_lz4
{
	/* 所有类型的接收器的通用信息。 */
	bbsink		base;

	/* 压缩级别。 */
	int			compresslevel;

	LZ4F_compressionContext_t ctx;
	LZ4F_preferences_t prefs;

	/* 输出缓冲区中分配的字节数。 */
	size_t		bytes_written;
} bbsink_lz4;

static void fc_bbsink_lz4_begin_backup(bbsink *fc_sink);
static void fc_bbsink_lz4_begin_archive(bbsink *fc_sink, const char *fc_archive_name);
static void fc_bbsink_lz4_archive_contents(bbsink *fc_sink, size_t fc_avail_in);
static void fc_bbsink_lz4_manifest_contents(bbsink *fc_sink, size_t fc_len);
static void fc_bbsink_lz4_end_archive(bbsink *fc_sink);
static void fc_bbsink_lz4_cleanup(bbsink *fc_sink);

static const bbsink_ops bbsink_lz4_ops = {
	.begin_backup = fc_bbsink_lz4_begin_backup,
	.begin_archive = fc_bbsink_lz4_begin_archive,
	.archive_contents = fc_bbsink_lz4_archive_contents,
	.end_archive = fc_bbsink_lz4_end_archive,
	.begin_manifest = bbsink_forward_begin_manifest,
	.manifest_contents = fc_bbsink_lz4_manifest_contents,
	.end_manifest = bbsink_forward_end_manifest,
	.end_backup = bbsink_forward_end_backup,
	.cleanup = fc_bbsink_lz4_cleanup
};
#endif

/*
 * 创建一个执行lz4压缩的新basebackup sink。
 */
bbsink * bbsink_lz4_new(bbsink *fc_next, pg_compress_specification *fc_compress)
{
#ifndef USE_LZ4
	ereport(ERROR,
			(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
			 errmsg("lz4 compression is not supported by this build")));
	return NULL;				/* 保持编译器安静 */
#else
	bbsink_lz4 *fc_sink;
	int			fc_compresslevel;

	Assert(fc_next != NULL);

	fc_compresslevel = fc_compress->level;
	Assert(fc_compresslevel >= 0 && fc_compresslevel <= 12);

	fc_sink = palloc0(sizeof(bbsink_lz4));
	*((const bbsink_ops **) &fc_sink->base.bbs_ops) = &bbsink_lz4_ops;
	fc_sink->base.bbs_next = fc_next;
	fc_sink->compresslevel = fc_compresslevel;

	return &fc_sink->base;
#endif
}

#ifdef USE_LZ4

/*
 * 开始备份。
 */
static void fc_bbsink_lz4_begin_backup(bbsink *fc_sink)
{
	bbsink_lz4 *fc_mysink = (bbsink_lz4 *) fc_sink;
	size_t		fc_output_buffer_bound;
	LZ4F_preferences_t *fc_prefs = &fc_mysink->prefs;

	/* 初始化压缩对象。 */
	memset(fc_prefs, 0, sizeof(LZ4F_preferences_t));
	fc_prefs->frameInfo.blockSizeID = LZ4F_max256KB;
	fc_prefs->compressionLevel = fc_mysink->compresslevel;

	/*
	 * 我们需要自己的缓冲区，因为我们将要传递给下一个接收器的数据
	 * 与传递给我们的数据不同。
	 */
	fc_mysink->base.bbs_buffer = palloc(fc_mysink->base.bbs_buffer_length);

	/*
	 * 由于LZ4F_compressUpdate()要求输出缓冲区的大小等于或
	 * 大于LZ4F_compressBound()，确保我们有下一个
	 * sink的bbs_buffer的长度可以容纳压缩输入
	 * 缓冲区。
	 */
	fc_output_buffer_bound = LZ4F_compressBound(fc_mysink->base.bbs_buffer_length,
											 &fc_mysink->prefs);

	/*
	 * 缓冲区长度预计应该是BLCKSZ的倍数，因此向上取整。
	 */
	fc_output_buffer_bound = fc_output_buffer_bound + BLCKSZ -
		(fc_output_buffer_bound % BLCKSZ);

	bbsink_begin_backup(fc_sink->bbs_next, fc_sink->bbs_state, fc_output_buffer_bound);
}

/*
 * 准备压缩下一个归档。
 */
static void fc_bbsink_lz4_begin_archive(bbsink *fc_sink, const char *fc_archive_name)
{
	bbsink_lz4 *fc_mysink = (bbsink_lz4 *) fc_sink;
	char	   *fc_lz4_archive_name;
	LZ4F_errorCode_t fc_ctxError;
	size_t		fc_headerSize;

	fc_ctxError = LZ4F_createCompressionContext(&fc_mysink->ctx, LZ4F_VERSION);
	if (LZ4F_isError(fc_ctxError))
		elog(ERROR, "could not create lz4 compression context: %s",
			 LZ4F_getErrorName(fc_ctxError));

	/* 首先将框架头写入目标缓冲区。 */
	fc_headerSize = LZ4F_compressBegin(fc_mysink->ctx,
									fc_mysink->base.bbs_next->bbs_buffer,
									fc_mysink->base.bbs_next->bbs_buffer_length,
									&fc_mysink->prefs);

	if (LZ4F_isError(fc_headerSize))
		elog(ERROR, "could not write lz4 header: %s",
			 LZ4F_getErrorName(fc_headerSize));

	/*
	 * 我们需要在输出缓冲区的头部之后写入压缩数据。
	 * 因此，请确保更新写入输出缓冲区的字节数的概念。
	 */
	fc_mysink->bytes_written += fc_headerSize;

	/* 在归档名称中添加“.lz4”。 */
	fc_lz4_archive_name = psprintf("%s.lz4", fc_archive_name);
	Assert(fc_sink->bbs_next != NULL);
	bbsink_begin_archive(fc_sink->bbs_next, fc_lz4_archive_name);
	pfree(fc_lz4_archive_name);
}

/*
 * 将输入数据压缩到输出缓冲区，直到输入
 * 数据耗尽。每当输出缓冲区低于输入缓冲区的压缩边界时，
 * 调用下一个sink的archive_contents()方法。
 *
 * 请注意，由于我们在压缩输入，因此
 * 我们很可能会消耗所有输入数据而不填满输出缓冲区。
 * 在这种情况下，当前输入数据的压缩表示不会
 * 实际发送到下一个bbsink，直到稍后的调用此函数，
 * 或者甚至可能在调用bbsink_lz4_end_archive()之前。
 */
static void fc_bbsink_lz4_archive_contents(bbsink *fc_sink, size_t fc_avail_in)
{
	bbsink_lz4 *fc_mysink = (bbsink_lz4 *) fc_sink;
	size_t		fc_compressedSize;
	size_t		fc_avail_in_bound;

	fc_avail_in_bound = LZ4F_compressBound(fc_avail_in, &fc_mysink->prefs);

	/*
	 * 如果可用字节的数量低于
	 * LZ4F_compressBound()计算的值，要求下一个sink处理数据，以便我们
	 * 可以清空缓冲区。
	 */
	if ((fc_mysink->base.bbs_next->bbs_buffer_length - fc_mysink->bytes_written) <
		fc_avail_in_bound)
	{
		bbsink_archive_contents(fc_sink->bbs_next, fc_mysink->bytes_written);
		fc_mysink->bytes_written = 0;
	}

	/*
	 * 压缩输入缓冲区并将其写入输出缓冲区。
	 */
	fc_compressedSize = LZ4F_compressUpdate(fc_mysink->ctx,
										 fc_mysink->base.bbs_next->bbs_buffer + fc_mysink->bytes_written,
										 fc_mysink->base.bbs_next->bbs_buffer_length - fc_mysink->bytes_written,
										 (uint8 *) fc_mysink->base.bbs_buffer,
										 fc_avail_in,
										 NULL);

	if (LZ4F_isError(fc_compressedSize))
		elog(ERROR, "could not compress data: %s",
			 LZ4F_getErrorName(fc_compressedSize));

	/*
	 * 更新我们已写入输出缓冲区的字节数的概念。
	 */
	fc_mysink->bytes_written += fc_compressedSize;
}

/*
 * lz4的内部缓冲区中可能还有一些数据；我们需要将
 * 这些数据刷新出来，并最终确定lz4帧，然后将其转发
 * 给后继sink作为归档内容。
 *
 * 然后我们可以结束对该归档的处理。
 */
static void fc_bbsink_lz4_end_archive(bbsink *fc_sink)
{
	bbsink_lz4 *fc_mysink = (bbsink_lz4 *) fc_sink;
	size_t		fc_compressedSize;
	size_t		fc_lz4_footer_bound;

	fc_lz4_footer_bound = LZ4F_compressBound(0, &fc_mysink->prefs);

	Assert(fc_mysink->base.bbs_next->bbs_buffer_length >= fc_lz4_footer_bound);

	if ((fc_mysink->base.bbs_next->bbs_buffer_length - fc_mysink->bytes_written) <
		fc_lz4_footer_bound)
	{
		bbsink_archive_contents(fc_sink->bbs_next, fc_mysink->bytes_written);
		fc_mysink->bytes_written = 0;
	}

	fc_compressedSize = LZ4F_compressEnd(fc_mysink->ctx,
									  fc_mysink->base.bbs_next->bbs_buffer + fc_mysink->bytes_written,
									  fc_mysink->base.bbs_next->bbs_buffer_length - fc_mysink->bytes_written,
									  NULL);

	if (LZ4F_isError(fc_compressedSize))
		elog(ERROR, "could not end lz4 compression: %s",
			 LZ4F_getErrorName(fc_compressedSize));

	/* 更新我们对已写字节数的概念。 */
	fc_mysink->bytes_written += fc_compressedSize;

	/* 发送我们已累积的输出字节。 */
	bbsink_archive_contents(fc_sink->bbs_next, fc_mysink->bytes_written);
	fc_mysink->bytes_written = 0;

	/* 释放资源。 */
	LZ4F_freeCompressionContext(fc_mysink->ctx);
	fc_mysink->ctx = NULL;

	/* 传递有关该归档已结束的信息。 */
	bbsink_forward_end_archive(fc_sink);
}

/*
 * 清单内容未压缩，但我们确实需要将其复制到
 * 后续接收器的缓冲区中，因为我们有自己的缓冲区。
 */
static void fc_bbsink_lz4_manifest_contents(bbsink *fc_sink, size_t fc_len)
{
	memcpy(fc_sink->bbs_next->bbs_buffer, fc_sink->bbs_buffer, fc_len);
	bbsink_manifest_contents(fc_sink->bbs_next, fc_len);
}

/*
 * 如果备份失败，请确保我们通过
 * 调用LZ4F_freeCompressionContext()来释放压缩上下文，以避免内存泄漏。
 */
static void fc_bbsink_lz4_cleanup(bbsink *fc_sink)
{
	bbsink_lz4 *fc_mysink = (bbsink_lz4 *) fc_sink;

	if (fc_mysink->ctx)
	{
		LZ4F_freeCompressionContext(fc_mysink->ctx);
		fc_mysink->ctx = NULL;
	}
}

#endif
